package com.fand.stater.listener;

import java.util.List;

import javax.annotation.Resource;

import cn.hutool.core.collection.CollUtil;
import com.fand.stater.annotation.MqSubscribe;
import com.fand.stater.constant.RocketTopicConstant;
import com.fand.stater.handler.SysUserOperateRecordHandler;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;

/**
 * Author: zhangjiabin
 * Email: javaboyok@163.com
 * CreateDate: 2021/6/23
 * Version: 1.0
 */
@Slf4j
@MqSubscribe(topic = RocketTopicConstant.SYS_USER_OPERATE_RECORD_TOPIC,tag = "SysUserESOperateRecordListenter")
public class SysUserESOperateRecordListenter implements MessageListenerConcurrently {

    @Resource
    private SysUserOperateRecordHandler sysUserOperateRecordHandler;

    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
        if (CollUtil.isEmpty(list)) {
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }
        list.stream().forEach(messageExt -> {
            String topic = messageExt.getTopic();
            String tags = messageExt.getTags();
            String record = new String(messageExt.getBody());
            log.info("SysUserESOperateRecordListenter consumeMessage;消费消息；topic:{};tags:{};body:{}", topic, tags, record);
            sysUserOperateRecordHandler.insertRecord2Es(record);
            log.info("消费消息成功");
        });
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }
}
